package com.yuyuan.thumb.config;

import org.apache.pulsar.client.api.BatchReceivePolicy;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.RedeliveryBackoff;
import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.pulsar.annotation.PulsarListenerConsumerBuilderCustomizer;

import java.util.concurrent.TimeUnit;

/**
 * 自定义 Customizer
 * Apache Pulsar 消费者配置，属于消息队列系统的可靠性处理核心配置。虽然代码中没有显式调用这些方法，但通过 Spring Boot 的自动装配机制，这些配置会被 Pulsar 客户端自动加载并应用。
 *
 * 配置批量处理消息数量
 *
 * 配置 NACK 重试策略
 * 配置 ACK 超时重试策略
 *
 * 配置死信处理策略
 *
 * 消息到达 先进行处理，若处理失败，根据NACK配置策略重试
 *                  若进程冻结，根据ACK超时重试策略进行重试
 *                  若重试超过N次，则发送到死信队列，继续失败则继续进入死信队列
 *                  若成功处理，则 ACK 确认消息已被消费
 *
 *  服务启动时加载 的，其配置在消费者初始化时一次性应用
 *
 * @author pine
 *
 * 该类使用了 @Configuration 注解，Spring Boot 在启动时会扫描并加载此类
 */
@Configuration
public class ThumbConsumerConfig<T> implements PulsarListenerConsumerBuilderCustomizer<T> {

    //将单条拉取优化为批量拉取,若 10 秒内未攒够 1000 条，立即发送已积累的消息

    // 当点赞系统突然遇到流量高峰（如热门内容被推荐）：
    //10秒内收到800条消息 → 立即批量消费（未达1000条但超时）
    //10秒内收到1500条消息 → 分2批消费（1000+500）
    //每秒只有5条消息 → 每10秒消费一次（约50条/批）
    //优势：减少数据库写入压力，提升吞吐量3-5倍

    @Override
    public void customize(ConsumerBuilder<T> consumerBuilder) {
        consumerBuilder.batchReceivePolicy(
                BatchReceivePolicy.builder()
                        // 每次处理 1000 条
                        // 单次最大拉取量
                        .maxNumMessages(1000)
                        // 设置超时时间（单位：毫秒）
                        .timeout(10000, TimeUnit.MILLISECONDS)
                        .build()
        );
    }

    // 配置 NACK 重试策略
    // NACK 重试策略（主动拒绝）
    //场景举例：
    //用户点赞时数据库死锁：
    //首次消费失败 → NACK → 1秒后重试
    //第二次仍遇死锁 → NACK → 2秒后重试
    //第三次成功 → ACK
    //设计价值：给数据库恢复时间，避免雪崩
    @Bean
    public RedeliveryBackoff negativeAckRedeliveryBackoff() {
        return MultiplierRedeliveryBackoff.builder()
                // 初始延迟 1 秒
                .minDelayMs(1000)
                // 最大延迟 60 秒
                .maxDelayMs(60000)
                // 每次重试延迟倍数
                .multiplier(2)
                .build();
    }

    // 配置 ACK 超时重试策略
    // ACK超时重试策略（超时未确认）
    // 重试序列：
    //5 → 15 → 45 → 135 → 300s（后续固定5min）
    //场景举例：
    //点赞消息处理时JVM发生Full GC：
    //消费线程卡顿30秒 → ACK超时
    //Broker 5秒后重投
    //若仍遇长时间GC → 15秒后再重投
    //特殊机制：
    //即使消费者崩溃，消息也会自动重投其他实例
    @Bean
    public RedeliveryBackoff ackTimeoutRedeliveryBackoff() {
        return MultiplierRedeliveryBackoff.builder()
                // 初始延迟 5 秒
                .minDelayMs(5000)
                // 最大延迟 300 秒
                .maxDelayMs(300_000)
                .multiplier(3)
                .build();
    }

    //配置死信处理机制
    //投递计数器：
    //初始消费=0，重试1=1，重试2=2，重试3=3 → 转死信
    //处理流程： 获取到消息之后进行消息消费
    //首次消费：JSON解析失败 → NACK
    //第二次：仍失败 → NACK
    //第三次：还是失败 → 转入 thumb-dlq-topic
    //原始消费者停止处理，死信队列触发告警
    @Bean
    public DeadLetterPolicy deadLetterPolicy() {
        return DeadLetterPolicy.builder()
                // 最大重试次数
                .maxRedeliverCount(3)
                // 死信主题名称
                .deadLetterTopic("thumb-dlq-topic")
                .build();
    }

}